package DataStreamApi.Transformation算子.合流;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;
import java.util.List;

/**
 * union的前提是同一种类型，一次可以合并多个。
 */
public class Flink01_UNION {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());


        List<Integer> list1 = Arrays.asList(
                1, 2, 3, 4, 5, 6, 7);


        List<Integer> list2 = Arrays.asList(
                3, 4, 2, 1, 5, 3, 5, 6, 7);

        List<String> list3 = Arrays.asList(
                "5", "6", "7");
        executionEnvironment.setParallelism(6);
        DataStreamSource<Integer> i1 = executionEnvironment.fromCollection(list1);
        DataStreamSource<Integer> i2 = executionEnvironment.fromCollection(list2);
        DataStreamSource<String> s1 = executionEnvironment.fromCollection(list3);


        DataStream<Integer> union = i1.union(i2);
        union.print("合并1");
        DataStream<Integer> union1 = union.union(s1.map(new MapFunction<String, Integer>() {
            @Override
            public Integer map(String value) throws Exception {
                return Integer.parseInt(value);
            }
        }));


        union1.print("合并2");

        executionEnvironment.execute();
    }
}
